//!@author denglibin
//!@date 2025/11/4 9:02
//!@desc

use crate::apps::tian_fu_mongo_doc::{TianFuDoc, TianFuIdDoc};
use crate::component::mongo::mongo_client::MongoClient;
use crate::dao::mysql::mysql_dao;

use crate::dao::mysql2sqlite;
use crate::sys::global::GlobalResult;
use chrono_tz::Tz;
use cron::Schedule;
use log::info;
use mongodb::bson::oid::ObjectId;
use mongodb::bson::{doc, Document};
use my_trait::MustHaveTableName;
use std::env;
use std::str::FromStr;
use std::time::Duration;
use tokio::time::Instant;

use my_util_lib::date;

const DB: &str = "sc_senti";
const COLLECTION_NAME: &str = "SH_D_ENTITYDATA";
///获取mongodb数据
/// ./my_tools mongodb://10.0.48.20:27017 tfrm_read 'tfrmPass123!@#' 'mysql://root:llvpimZdw3%y@localhost:3306/tian_fu_data'
/// ./my_tools mongodb://192.168.6.170:27017 mongo mongo  'mysql://denglibin:denglibin@192.168.6.9:3308/denglibin'
pub async fn get_data() -> Result<(), Box<dyn std::error::Error>> {
    let args = env::args().collect::<Vec<String>>();
    if args.len() < 5 {
        log::error!("Missing arguments!, uri, username, password, mysql_url");
        return Ok(());
    }
    let uri: String = args[1].clone();
    let username: String = args[2].clone();
    let password: String = args[3].clone();
    let mysql_url = &args[4]; // "mysql://root:denglibin@192.168.6.9:3308/denglibin";

    let client = MongoClient::new(&uri, Some(username), Some(password)).await?;

    show_db_collections(&client).await?;
    // count(&client).await?;

    //直接 采集
    // to_mysql(&client, mysql_url).await?;
    //定时采集
    schedule_get_data(&client, mysql_url).await?;

    Ok(())
}

/// 定时采集
pub async fn schedule_get_data(
    client: &MongoClient,
    mysql_url: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    //定时执行
    //cron 表达式
    let cron_expression = "0 0 3 * * ?"; //每天03点
    let schedule = Schedule::from_str(cron_expression)?;
    loop {
        let next_execution: Option<chrono::prelude::DateTime<Tz>> =
            schedule.upcoming(Tz::Asia__Shanghai).next();
        if let Some(next_time) = next_execution {
            info!("下个执行点:{}", next_time);
            let timestamp = next_time.timestamp_millis() as u64;
            let now_stamp = date::get_sys_timestamp_millis() as u64;
            let op_i = Instant::now().checked_add(Duration::from_millis(timestamp - now_stamp));
            if let Some(i) = op_i {
                tokio::time::sleep_until(i).await;
                //时间到了，采集
                to_mysql(&client, mysql_url).await?;
            }
        }
    }
}

///查询库和库中的集合
pub async fn show_db_collections(client: &MongoClient) -> Result<(), Box<dyn std::error::Error>> {
    let db_names = client.list_database_names().await?;
    log::info!("db_names:{:?}", db_names);
    for db_name in db_names {
        let collection_names_r = client.list_collection_names(db_name.as_str()).await;
        if let Ok(collection_names) = collection_names_r {
            log::info!("db:{}, collection_names:{:?}", db_name, collection_names);
        } else {
            log::error!("db:{},{}", db_name, collection_names_r.unwrap_err());
        }
    }

    Ok(())
}

///统计数量
pub async fn count(client: &MongoClient) -> Result<(), Box<dyn std::error::Error>> {
    let c = client.count_by_metadata(DB, COLLECTION_NAME).await?;
    log::info!("db:{}, collection:{}, count:{}", DB, COLLECTION_NAME, c);
    Ok(())
}

///获取示例数据
pub async fn to_mysql(
    client: &MongoClient,
    mysql_url: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let pool = mysql_dao::get_conn_pool(mysql_url, 1).await?;
    let table_name = TianFuDoc::table_name();
    log::info!("table_name:{}", table_name);
    let mut start_id = "0";

    //获取最后一条id
    let sql = format!(
        "select id, doc_id from {} order by id desc limit 1",
        table_name
    );
    let last = mysql_dao::select_list(&sql, &pool).await?;

    if !last.1.is_empty() {
        start_id = last.1.get(0).unwrap().get(1).unwrap().as_str().unwrap();
    }
    log::info!("start_id:{}", start_id);

    let total = client.count_by_metadata(DB, COLLECTION_NAME).await?;

    let oid_r = ObjectId::parse_str(start_id);

    log::info!("start_id:{:?}, total:{}", start_id, total);
    let mut query = if let Ok(oid) = oid_r {
        doc! {"_id":{ "$gt": oid }}
    } else {
        doc! {}
    };
    log::info!("query:{:?}", query);

    let mut count = 0;
    let mut docs = vec![];
    let mut doc_count = 0;
    loop {
        client
            .query_list::<Document, _>(DB, COLLECTION_NAME, query.clone(), |doc| {
                let id_r = doc.get_object_id("_id");
                let mut current_id = "".to_string();
                if let Ok(id) = id_r {
                    query = doc! {"_id":{ "$gt": id }};
                    current_id = id.to_string();
                    let data_region_r = doc.get_document("dataregion");
                    if let Ok(region) = data_region_r {
                        let publish_status_r = region.get_str("publishstatus");
                        if let Ok(publish_status) = publish_status_r {
                            if publish_status == "上架" {
                                let note_r = region.get_str("note");
                                if let Ok(note) = note_r {
                                    let publish_date_r = region.get_datetime("publishdate");
                                    //发布时间
                                    let publish_date = if let Ok(time) = publish_date_r {
                                        Some(time.timestamp_millis())
                                    } else {
                                        None
                                    };
                                    //来源
                                    let original_r = region.get_str("original");
                                    let original = if let Ok(o) = original_r {
                                        Some(o.to_string())
                                    } else {
                                        None
                                    };
                                    let system_name_r = region.get_str("systemname");
                                    let system_name = if let Ok(o) = system_name_r {
                                        Some(o.to_string())
                                    } else {
                                        None
                                    };

                                    let title_r = region.get_str("title");
                                    if let Ok(title) = title_r {
                                        let tian_fu_doc = TianFuDoc {
                                            doc_id: current_id.to_string(),
                                            title: title.to_string(),
                                            content: note.to_string(),
                                            publish_date,
                                            original,
                                            system_name,
                                        };
                                        doc_count += 1;
                                        docs.push(tian_fu_doc);
                                    }
                                }
                            }
                        }
                    }
                }

                count += 1;
                if count % 1000 == 0 {
                    log::info!("进度:{}/{}, id={}", count, total, current_id);
                }
                if docs.len() >= 100 {
                    log::info!("doc数量:{}", doc_count);
                    return false; //本次查询结束
                }

                true
            })
            .await?;
        if docs.is_empty() {
            log::info!("未获取到数据，任务结束");
            break;
        }

        let insert_r = mysql_dao::insert_batch::<TianFuDoc>(&docs, &pool).await;
        if let Err(e) = insert_r {
            log::error!("mysql_dao::insert_batch error:{:?}", e);
            break;
        }
        docs.clear(); //清空
    }

    Ok(())
}

//mysql存到sqlite中
pub async fn mysql2sqlite() -> GlobalResult<()> {
    let args = env::args().collect::<Vec<String>>();
    if args.len() < 2 {
        log::error!("Missing arguments!,mysql_url");
        return Ok(());
    }
    let mysql_url = &args[1]; //"mysql://root:llvpimZdw3%y@localhost:3306/tian_fu_data";
    let sqlite_url = "sqlite:tian_fu_doc.db";
    let columns = vec![
        "id".to_string(),
        "doc_id".to_string(),
        "title".to_string(),
        "content".to_string(),
    ];
    let table_name = "tian_fu_doc";
    let create_table_sql =
        "create table if not exists tian_fu_doc(id integer primary_key, doc_id text, title text, content text)";
    mysql2sqlite::mysql2sqlite::<TianFuIdDoc>(
        mysql_url,
        sqlite_url,
        table_name,
        create_table_sql,
        columns,
        0,
    )
    .await
}
